diff --git a/Sources/LanguageServerProtocolJSONRPC/JSONRPCConnection.swift b/Sources/LanguageServerProtocolJSONRPC/JSONRPCConnection.swift
index c2c1c696..8fb2e491 100644
--- a/Sources/LanguageServerProtocolJSONRPC/JSONRPCConnection.swift
+++ b/Sources/LanguageServerProtocolJSONRPC/JSONRPCConnection.swift
@@ -45,21 +45,17 @@ public final class JSONRPCConnection: Connection {
 
   /// The message handler that handles requests and notifications sent through this connection.
   ///
-  /// Access to this must be be guaranteed to be sequential to avoid data races. Currently, all access are
-  ///  - `init`: Reference to `JSONRPCConnection` trivially can't have escaped to other isolation domains yet.
-  ///  - `start`: Is required to be call in the same serial code region as the initializer, so
-  ///    `JSONRPCConnection` can't have escaped to other isolation domains yet.
-  ///  - `deinit`: Can also only trivially be called once.
+  /// - Important: Must only be accessed from `queue`, `init` and `deinit` to avoid data races.
   nonisolated(unsafe) private var receiveHandler: MessageHandler?
 
-  /// The queue on which we read the data
+  /// The queue on which we read and send the data
   private let queue: DispatchQueue = DispatchQueue(label: "jsonrpc-queue", qos: .userInitiated)
 
-  /// The queue on which we send data.
-  private let sendQueue: DispatchQueue = DispatchQueue(label: "jsonrpc-send-queue", qos: .userInitiated)
+  /// The file handle with which the other process sends data to us.
+  private let receiveFD: FileHandle
+  /// The file handle with which we can send data to the other process
+  private let sendFD: FileHandle
 
-  private let receiveIO: DispatchIO
-  private let sendIO: DispatchIO
   private let messageRegistry: MessageRegistry
 
   /// If non-nil, all input received by this `JSONRPCConnection` will be written to the file handle
@@ -86,7 +82,6 @@ public final class JSONRPCConnection: Connection {
   /// Buffer of received bytes that haven't been parsed.
   ///
   /// Access to this must be be guaranteed to be sequential to avoid data races. Currently, all access are
-  ///  - The `receiveIO` handler: This is synchronized on `queue`.
   ///  - `requestBufferIsEmpty`: Also synchronized on `queue`.
   private nonisolated(unsafe) var requestBuffer: [UInt8] = []
 
@@ -136,64 +131,8 @@ public final class JSONRPCConnection: Connection {
     state = .created
     self.messageRegistry = messageRegistry
 
-    let ioGroup = DispatchGroup()
-
-    #if os(Windows)
-    let rawInFD = dispatch_fd_t(bitPattern: inFD._handle)
-    #else
-    let rawInFD = inFD.fileDescriptor
-    #endif
-
-    ioGroup.enter()
-    receiveIO = DispatchIO(
-      type: .stream,
-      fileDescriptor: rawInFD,
-      queue: queue,
-      cleanupHandler: { (error: Int32) in
-        if error != 0 {
-          logger.fault("IO error \(error)")
-        }
-        ioGroup.leave()
-      }
-    )
-
-    #if os(Windows)
-    let rawOutFD = dispatch_fd_t(bitPattern: outFD._handle)
-    #else
-    let rawOutFD = outFD.fileDescriptor
-    #endif
-
-    ioGroup.enter()
-    sendIO = DispatchIO(
-      type: .stream,
-      fileDescriptor: rawOutFD,
-      queue: sendQueue,
-      cleanupHandler: { (error: Int32) in
-        if error != 0 {
-          logger.fault("IO error \(error)")
-        }
-        ioGroup.leave()
-      }
-    )
-
-    ioGroup.notify(queue: queue) { [weak self] in
-      guard let self else { return }
-      for outstandingRequest in self.outstandingRequests.values {
-        outstandingRequest.replyHandler(LSPResult.failure(ResponseError.internalError("JSON-RPC Connection closed")))
-      }
-      self.outstandingRequests = [:]
-      self.receiveHandler = nil  // break retain cycle
-      Task {
-        await self.closeHandler?()
-      }
-    }
-
-    // We cannot assume the client will send us bytes in packets of any particular size, so set the lower limit to 1.
-    receiveIO.setLimit(lowWater: 1)
-    receiveIO.setLimit(highWater: Int.max)
-
-    sendIO.setLimit(lowWater: 1)
-    sendIO.setLimit(highWater: Int.max)
+    self.receiveFD = inFD
+    self.sendFD = outFD
   }
 
   /// Creates and starts a `JSONRPCConnection` that connects to a subprocess launched with the specified arguments.
@@ -293,27 +232,20 @@ public final class JSONRPCConnection: Connection {
       state = .running
       self.receiveHandler = receiveHandler
       self.closeHandler = closeHandler
+    }
 
-      receiveIO.read(offset: 0, length: Int.max, queue: queue) { done, data, errorCode in
-        guard errorCode == 0 else {
-          #if !os(Windows)
-          if errorCode != POSIXError.ECANCELED.rawValue {
-            logger.fault("IO error reading \(errorCode)")
-          }
-          #endif
-          if done { self.closeAssumingOnQueue() }
-          return
-        }
-
-        if done {
-          self.closeAssumingOnQueue()
-          return
-        }
-
-        guard let data = data, !data.isEmpty else {
-          return
-        }
+    self.receiveFD.readabilityHandler = { fileHandle in
+      let data = orLog("Reading from \(self.name)") { try fileHandle.read(upToCount: 1) } ?? Data()
+      guard !data.isEmpty else {
+        // We have reached the end of `receiveFD`, close the connection. This will also set the `readabilityHandler` of
+        // `receiveFD` to `nil`, breaking the retain cycle.
+        self.close()
+        return
+      }
 
+      // We can run the actual data handling asynchronously in the background, all we need to ensure is that the data is
+      // handled in-order, which the queue does.
+      self.queue.async {
         orLog("Writing input mirror file") {
           try self.inputMirrorFile?.write(contentsOf: data)
         }
@@ -554,17 +486,17 @@ public final class JSONRPCConnection: Connection {
     orLog("Writing output mirror file") {
       try outputMirrorFile?.write(contentsOf: dispatchData)
     }
-    sendIO.write(offset: 0, data: dispatchData, queue: sendQueue) { [weak self] done, _, errorCode in
-      if errorCode != 0 {
-        logger.fault("IO error sending message \(errorCode)")
-        if done, let self {
-          // An unrecoverable error occurs on the channel’s file descriptor.
-          // Close the connection.
-          self.queue.async {
-            self.closeAssumingOnQueue()
-          }
-        }
+    do {
+      try sendFD.write(contentsOf: dispatchData)
+    } catch {
+      logger.fault("IO error sending message \(error.forLogging)")
+      // Match the pattern of `close()` but call `closeAssumingOnQueue` asynchronously to make sure that
+      // `closeAssumingOnQueue` is executed after all data from `receiveFD` has been read.
+      self.receiveFD.readabilityHandler = nil
+      self.queue.async {
+        self.closeAssumingOnQueue()
       }
+      return
     }
   }
 
@@ -584,7 +516,7 @@ public final class JSONRPCConnection: Connection {
   /// If an unrecoverable error occurred on the channel's file descriptor, the connection gets closed.
   ///
   /// - Important: Must be called on `queue`
-  func send(_ message: JSONRPCMessage) {
+  private func send(_ message: JSONRPCMessage) {
     dispatchPrecondition(condition: .onQueue(queue))
 
     let encoder = JSONEncoder()
@@ -646,7 +578,12 @@ public final class JSONRPCConnection: Connection {
   /// The user-provided close handler will be called *asynchronously* when all outstanding I/O
   /// operations have completed. No new I/O will be accepted after `close` returns.
   public func close() {
-    queue.sync { closeAssumingOnQueue() }
+    // Stop reading any more data from `receiveFD`. Scheduling `closeAssumingOnQueue` on `queue` after closing
+    // `receiveFD` ensures that we won't read any more data after `closeAssumingOnQueue`.
+    self.receiveFD.readabilityHandler = nil
+    queue.sync {
+      closeAssumingOnQueue()
+    }
   }
 
   /// Close the connection, assuming that the code is already executing on `queue`.
@@ -654,15 +591,25 @@ public final class JSONRPCConnection: Connection {
   /// - Important: Must be called on `queue`.
   private func closeAssumingOnQueue() {
     dispatchPrecondition(condition: .onQueue(queue))
-    sendQueue.sync {
-      guard state == .running else { return }
-      state = .closed
-
-      logger.log("Closing JSONRPCConnection...")
-      // Attempt to close the reader immediately; we do not need to accept remaining inputs.
-      receiveIO.close(flags: .stop)
-      // Close the writer after it finishes outstanding work.
-      sendIO.close()
+    // Stop receiving any more
+    guard state == .running else { return }
+    state = .closed
+
+    logger.log("Closing JSONRPCConnection...")
+    // Attempt to close the reader immediately; we do not need to accept remaining inputs.
+    // Close the writer after it finishes outstanding work.
+    do {
+      try sendFD.close()
+    } catch {
+      logger.error("Failed to close outFD: \(error.forLogging)")
+    }
+    for outstandingRequest in self.outstandingRequests.values {
+      outstandingRequest.replyHandler(LSPResult.failure(ResponseError.internalError("JSON-RPC Connection closed")))
+    }
+    self.outstandingRequests = [:]
+    self.receiveHandler = nil  // break retain cycle
+    Task {
+      await self.closeHandler?()
     }
   }
 
@@ -694,13 +641,13 @@ public final class JSONRPCConnection: Connection {
     id: RequestID,
     reply: @escaping @Sendable (LSPResult<Request.Response>) -> Void
   ) {
-    self.queue.sync {
-      guard readyToSend() else {
+    self.queue.async {
+      guard self.readyToSend() else {
         reply(.failure(.serverCancelled))
         return
       }
 
-      outstandingRequests[id] = OutstandingRequest(
+      self.outstandingRequests[id] = OutstandingRequest(
         responseType: Request.Response.self,
         replyHandler: { anyResult in
           let result = anyResult.map { $0 as! Request.Response }
@@ -732,7 +679,7 @@ public final class JSONRPCConnection: Connection {
         """
       )
 
-      send(.request(request, id: id))
+      self.send(.request(request, id: id))
       return
     }
   }